package com.bigdata

import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import com.bigdata.common.constants.Constants
import com.bigdata.common.utils.KafkaCommonUtils
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

/**
 * 模拟生成登陆事件日志
 */
object KafkaProducerLog {

  def main(args: Array[String]): Unit = {
    writeToKafka("loginEvent")
//    test()
  }


  def writeToKafka(topic: String): Unit = {
    val producer = KafkaCommonUtils.createKafkaProducer(topic)
    try {
      while (true) {
        val userId = new Random().nextInt(10)
        val ip = s"${new Random().nextInt(255)}.${new Random().nextInt(255)}.${new Random().nextInt(255)}.${new Random().nextInt(255)}"
        val eventType = if (new Random().nextBoolean()) "success" else "fail"
        val eventTime = System.currentTimeMillis() - new Random().nextInt(1000000)
        val log = s"$userId,$ip,$eventType,$eventTime"
        val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
        val timeStr: String = simpleDateFormat.format(new Date())
        println(s"$timeStr: [$log]")
        producer.send(new ProducerRecord[String, String](topic, log))
        Thread.sleep(500)
      }
    } catch {
      case ex: Exception => ex.printStackTrace()
    }
    finally {
      producer.close()
    }

  }

  def test(): Unit ={
    KafkaCommonUtils.produceMockData(Constants.Topic_Login){
      val userId = new Random().nextInt(10)
      val ip = s"${new Random().nextInt(255)}.${new Random().nextInt(255)}.${new Random().nextInt(255)}.${new Random().nextInt(255)}"
      val eventType = if (new Random().nextBoolean()) "success" else "fail"
      val eventTime = System.currentTimeMillis() - new Random().nextInt(1000000)
      val log = s"$userId,$ip,$eventType,$eventTime"
      println(log)
      log
    }
  }

}